package com.aididi.demospringbootrabbitmqcallback.Producer;

import com.aididi.demospringbootrabbitmqcallback.constant.Constants;
import com.aididi.demospringbootrabbitmqcallback.dao.BrokerMessageLogMapper;
import com.aididi.demospringbootrabbitmqcallback.entity.BrokerMessageLogVO;
import com.aididi.demospringbootrabbitmqcallback.entity.Order;
import com.aididi.demospringbootrabbitmqcallback.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @Auther: xuzexiang
 * @Date: 2018-11-18 0018 23:27
 * @Description:    订单消息发送者
 */
@Component
@Slf4j
public class OrderSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private BrokerMessageLogMapper brokerMessageLogMapper;
    /**
     * 创建一个消息是否投递成功的回调方法
     * 回调方法：confirm确认发送订单   ,确认消息是否发送成功
     */
    private final RabbitTemplate.ConfirmCallback  callback  = new RabbitTemplate.ConfirmCallback() {
        /**
         *
         * @param correlationData 消息的附加信息
         * @param ack true for ack, false for nack
         * @param cause 是一个可选的原因，对于nack，如果可用，否则为空。
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            log.info("消息标识唯一id  correlationData:"+correlationData + "---它的id是"+correlationData.getId());
            String messageId =  correlationData.getId();
            if(ack){
                //如果confirm返回成功，则进行更新
                BrokerMessageLogVO brokerMessageLogVO = new BrokerMessageLogVO();
                brokerMessageLogVO.setMessageId(messageId);
                //设置状态为1，代表消息发送成功
                brokerMessageLogVO.setStatus(Constants.OrderSendStatus.SEND_SUCCESS);
                //
                brokerMessageLogMapper.changeBrokerMessageLogStatus(brokerMessageLogVO);
            }else {
                // 失败则进行具体的后续操作：重试或者补偿等
                log.error("异常处理...");
                log.error("ack,消息投递exchange失败，msgId:{},原因{}" ,messageId, cause);
            }
        }
    };

    /**
     * 回调函数: return返回 ，  消息失败返回，比如路由不到队列时触发此回调
     * exchange到queue成功,则不回调return
     * exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
     */
    final RabbitTemplate.ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText,
                                    String exchange, String routingKey) {
            log.error("消息丢失:{}, exchange: {},routeKey: {},replyCode: {},replyText: {}",message, exchange, routingKey, replyCode, replyText);
        }
    };

    /**
     * 扩展点，在消息转换完成之后，发送之前调用；可以修改消息属性、消息头信息
     */
    private final MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
        @Override
        public Message postProcessMessage(Message message) throws AmqpException {
            MessageProperties properties = message.getMessageProperties();
            /**
             * 设置消息的优先级
             */
            properties.setPriority(9);
            /**
             * 设置消息发送到队列中的模式，持久化|非持久化（只存在于内存中）
             */
            properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            /**
             * 消息的过期时间 ,5秒之后过期
             */
            //properties.setExpiration("5000");
            /**
             * Set the x-delay header.  延迟时间
             */
            //properties.setDelay(10000);
            /**
             * 或设置延迟header
             */
            //properties.getHeaders().put("x-delay", 10000);

            return message;
        }
    };


    /**
     *  发送订单
     * @param order
     */
    public void send(Order order){
        /**
         * 设置生产者消息publish-confirm回调函数
         */
        rabbitTemplate.setConfirmCallback(callback);
        /**
         * 设置消息退回回调函数
         */
        rabbitTemplate.setReturnCallback(returnCallback);
        /**
         * 新增消息转换完成后、发送之前的扩展点
         */
        rabbitTemplate.setBeforePublishPostProcessors(messagePostProcessor);
        CorrelationData correlationData = new CorrelationData(order.getMessageId());
        //exchange交换机
        //Routing Key
        //消息体内容
        //correlationData 消息唯一id
        rabbitTemplate.convertAndSend("order-exchange","order.a", JsonUtil.convertObjectToJson(order),correlationData);
        //延迟消息
/*
        rabbitTemplate.convertAndSend("order-exchange","order.a", JsonUtil.convertObjectToJson(order),
                postProcessor ->{
                    // 设置延迟毫秒值
                    postProcessor.getMessageProperties().setExpiration(String.valueOf(120000));//在这里也可以设置超时时间,也可以设置x-message-ttl
                    postProcessor.getMessageProperties().setDelay(String.valueOf(120000));//需要利用rabbitmq_delayed_message_exchange插件设置延迟消息
                    return postProcessor;
                }, correlationData);
*/

    }
}
